package com.xuchaotong.zk.utils;

import com.xuchaotong.zk.listener.ZookeeperConnectionListener;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author 311633
 * @version 1.0
 * @ClassName:
 * @Description:
 * @date 2018/2/1 11:00
 */
public class ZookeeperUtils {
    private static final Logger LOG = LoggerFactory.getLogger(ZookeeperUtils.class);

    private static final String connectionInfo = "10.230.34.217:2181,10.230.34.217:2182,10.230.34.217:2183";
    private static ZookeeperUtils instance = new ZookeeperUtils();
    private static Object lock = new Object();

    private static CuratorFramework client;

    private ZookeeperUtils() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.builder()
                .connectString(connectionInfo)
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(5000)
                .retryPolicy(retryPolicy)
                .build();


        client.start();
    }

    public static ZookeeperUtils getInstance(){
        if(instance == null){
            synchronized (lock){
                instance = new ZookeeperUtils();
            }
        }
        return instance;
    }

    public CuratorFramework getClient(){
        return client;
    }

    public void setNodeCacheListener(String path, boolean dataIsCompressed) {
        try {
            NodeCache nodeCache = new NodeCache(client, path, dataIsCompressed);
            NodeCacheListener nodeCacheListener = new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    ChildData childData = nodeCache.getCurrentData();
                    LOG.info("ZNode节点状态改变, path={}", childData.getPath());
                    LOG.info("ZNode节点状态改变, data={}", new String(childData.getData()));
                    LOG.info("ZNode节点状态改变, stat={}", childData.getStat());
                }
            };
            nodeCache.getListenable().addListener(nodeCacheListener);
            nodeCache.start();
        } catch (Exception e) {
            LOG.error("创建NodeCache监听失败, path={}", path);
        }
    }

    public void addConnectionStateListener(ConnectionStateListener listener){
        client.getConnectionStateListenable().addListener(listener);
    }
}
